package com.atguigu.gmall.apps

import java.time.format.DateTimeFormatter
import java.time.{Instant, LocalDate, LocalDateTime, ZoneId}
import java.util

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.Beans.{CouponAlertInfo, EventLog}
import com.atguigu.gmall.utils.Constant.{ESConstant, KafkaConstant}
import com.atguigu.gmall.utils.{MyESUtil, MykafkaUtil}
import io.searchbox.client.JestClient
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.Minutes
import org.apache.spark.streaming.dstream.{DStream, InputDStream}

/**
  * @ClassName: AlterApp
  * @Description:
  * @Author: kele
  * @Date: 2021/3/25 11:22
  *
  * 预警日志：
  *   1、消费kafka数据，将数据转化为EventLog类
  *   2、开窗，开窗长度为5min，统计5min中的数据
  *   3、按照mid进行聚合
  *   4、过滤：去除掉5min，同一设备，登录三个账号，没有点击操作
  *
  *
  **/
object AlterApp extends baseApp {
  override val AppName: String = "Alter"
  override val time: Int = 5

  def main(args: Array[String]): Unit = {

    runApp({

      val ds: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(KafkaConstant.KAFKA_TOPIC_EVENT,streamingContext)

      //1、获取event数据，给logDate、logHour赋值将数据封装为EventLog类，
      val ds1: DStream[EventLog] = ds.map(record => {

        val eventLog: EventLog = JSON.parseObject(record.value(), classOf[EventLog])

        var time: LocalDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(eventLog.ts), ZoneId.of("Asia/Shanghai"))

        val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")

        eventLog.logDate = time.format(formatter)
        eventLog.logHour = time.getHour.toString

        eventLog

      })

      //2、开窗，统计5min内的数据
      val ds2: DStream[(String, Iterable[EventLog])] = ds1.window(Minutes(5))
        //3、按照mid进行分组，
        .map(event => (event.mid, event))
        .groupByKey()

      //4、数据封装为预警日志格式，并且过滤
      val ds3: DStream[CouponAlertInfo] = ds2.map {

        //返回为：需要预警的返回CouponAlertInfo，不需要预警的返回null
        case (mid, eventlogs) => {

          //mid上登录的用户
          var uids:util.HashSet[String] = new util.HashSet[String]()
          //mid上领域优惠券
          var itemIds:util.HashSet[String] = new util.HashSet[String]()
          //mid的操作
          var events:util.List[String] = new util.ArrayList[String]()

          var flag: Boolean = true

          //①、遍历每个rdd,按照预警日志的格式输出,将数据封装起来
          eventlogs.foreach(eventlog => {

            //②、将每个操作封装
            events.add(eventlog.evid)

            //有点击商品的不需要预警
            if (eventlog.evid.equals("clickItem")) {
              flag = false
            }

            //保存领券的用户和领的券
            if (eventlog.evid.equals("coupon")) {

              uids.add(eventlog.uid)
              itemIds.add(eventlog.itemid)
            }
          })

          //如果点击了商品，或者领取 优惠券的用户小于3，则不保存预警日志，返回null
          if (flag == false || uids.size() < 3) {
            //返回null
            null
          } else {
            //如果到达规定的预警条件，则返回相关的数据
            CouponAlertInfo(mid, uids, itemIds, events, System.currentTimeMillis())
          }
        }
      }

      //将非空的数据过滤掉
      val ds4: DStream[CouponAlertInfo] = ds3.filter(_!=null)


      //将数据保存到ES上,每分钟记录一次
      ds4.foreachRDD(rdd=>{

        rdd.foreachPartition(coup=>{

          //获取连接
          //val client: JestClient = MyESUtil.getClient

          //将每天的数据作为一个整体,
          //这里需要借助模板，所以要加前缀
          var indexName:String = ESConstant.ES_ALTER_INFO+LocalDate.now().toString

          val doc: List[(String, CouponAlertInfo)] = coup.map(doc => {

            //每个设备每分钟统计一次
            val time: String = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm"))

            //第一个参数作为id，第二个是数据
            (doc.mid + "_" + time, doc)

          }).toList


          //将数据写入到ES中
          MyESUtil.insertBulk(indexName,doc)

          //关闭连接
          //MyESUtil.close(client)

        })

      })

    })
  }

}
